package com.源码分析.以编程方式创建流;

import com.domain.User;

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

public class CreateFluxSink {

	private FluxSink<User> userSink;
	private Flux<User> flux;

	public CreateFluxSink() {
		// 利用FluxSink可以构建出一个用于持续生成User对象的Flux对象
		flux = Flux.<User>create(sink -> {
			this.userSink = sink;
		}).publish().autoConnect(); // 见冷热序列，使用publish()和autoConnect()方法确保一旦FluxSink产生数据，Flux就准备随时进行发送
		flux.subscribe(System.out::println); // 第一次调用subscribe()让流启动，但此时还没有数据，直到调用publishUser()，从而触发userSink.next(user)放入了数据
	}

	private Mono<Void> publishUser(String id, String name){
		User user = new User(id, name);
		Mono<User> monoUser = Mono.just(user);
		return monoUser.map(event -> {
			user.setPhone("138xxxxxx");
			return userSink.next(user);
		}).then();
	}

	public static void main(String[] args) {
		CreateFluxSink sink = new CreateFluxSink();
		Mono<Void> re = sink.publishUser("1", "zzz");
		re.subscribe(); // 目的是触发Mono<Void>的订阅
	}

}
